package com.feidee.fdhadoop.hbase;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.util.ToolUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 清空 HBase表中的指定列
 *     如果设置了默认值, 则将该列置为默认值(默认值支持空字符串);
 *     否则, 则将删除该列
 */
public class HBaseTableCleaner {

    protected static Logger logger = Logger.getLogger(HBaseTableCleaner.class);

    public static void main(String[] args) {
        if (args == null || args.length < 3 || args.length > 4) {
            logger.error("### args error!");
            System.exit(-1);
        }
        String tableName = args[0];
        String family = args[1];
        String columns = args[2];
        if(!ToolUtils.isNotNull(tableName) || !ToolUtils.isNotNull(family) || !ToolUtils.isNotNull(columns)){
            logger.error("args[0]:tableName,args[1]:family,args[2]:columns");
        }
        String defaultv = null;
        if (args.length == 4) {
            defaultv = args[3];
        }
        logger.info("开始清空操作...");
        cleanUp(tableName, family, columns.split(","), defaultv);
    }

    private static void cleanUp(String tableName, String family, String[] columns, String defaultv) {
        try {
            if (HBaseManager.tableExists(tableName)) {
                Table hTable = HBaseManager.getHTable(tableName);
                byte[][] prefixes = new byte[columns.length][];
                for(int i = 0; i < columns.length; i++){
                    prefixes[i] = Bytes.toBytes(columns[i]);
                }
                Filter filter = new MultipleColumnPrefixFilter(prefixes);//列过滤器
                Scan scan = new Scan();
                scan.addFamily(Bytes.toBytes(family));
                scan.setFilter(filter);
                ResultScanner rs = hTable.getScanner(scan);
                int count = 0;
                if (rs != null) {
                    List<Map<String, String>> setList = new ArrayList<>();
                    List<List<String>>  delList = new ArrayList<>();
                    for (Result result : rs) {
                        Cell[] cells = result != null ? result.rawCells() : null;
                        if (cells != null && cells.length > 0) {
                            Map<String, String> map = new HashMap<String, String>();
                            List<String> list = new ArrayList<>();
                            for (Cell cell : cells) {
                                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                                // 如果该列有默认值, 则设置默认值
                                if (defaultv != null) {
                                    map.put("rowkey", rowkey);
                                    map.put(qualifier, defaultv);
                                    setList.add(map);
                                }else{
                                    list.add(rowkey);
                                    list.add(qualifier);
                                    delList.add(list);
                                }
                            }
                            if(setList.size() >= 5000){//批量设置默认值
                                HBaseManager.addRecordsBatch(tableName, family, setList);
                                setList.clear();
                            }
                            if(delList.size() >= 5000){//批量清空
                                count += delList.size();
                                logger.info("删除数据"+count);
                                HBaseManager.deleteColumnBatch(tableName, family, delList);
                                delList.clear();
                            }
                        }
                    }
                } else {
                    logger.error("### 数据为空, 请检查列族、列名是否正确!");
                }
                logger.info("数据清理完毕");
            } else {
                logger.error(tableName + " is not exists!");
            }
        } catch (IOException e) {
            logger.error("### HBase异常!", e);
        } finally {
            HBaseManager.closeConnection();
        }
    }

}
